package realtime.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;

/**
 * @Create 2023/9/5 19:52
 */
public class HbaseUtil {

    /**
     * 获取hbase链接
     * @return
     */
    public static Connection getHbaseConnection(){
        Configuration conf = HBaseConfiguration.create();
        //配置zookeeper
        conf.set("hbase.zookeeper.quorum","localhost");
        conf.set("hbase.zookeeper.property.clientPort","12181");

        try {
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            System.out.println("====== 获取hbase链接失败 ========");
            throw new RuntimeException(e);
        }
    }

    /**
     * 关闭hbase链接
     * @param conn
     */
    public static void closeHbaseConnection(Connection conn){
        if (conn != null&& !conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                System.out.println("链接关闭失败");
                e.printStackTrace();
            }
        }
    }

    /**
     * 删除hbase表
     * @param conn 链接
     * @param namespace 命名空间
     * @param tableName 表名
     */
    public static void dropTable(Connection conn,String namespace,String tableName){
        Admin admin;

        try {
            admin = conn.getAdmin();

            TableName tableNameHbase = TableName.valueOf(namespace, tableName);
            if (!admin.tableExists(tableNameHbase)){
                System.out.println("要删除的表: "+namespace+":"+tableName+"不存在。。。");
                return;
            }
            //先禁用再删除
            admin.disableTable(tableNameHbase);
            admin.deleteTable(tableNameHbase);

        } catch (IOException e) {
            System.out.println("获取HbaseAdmin失败");
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建Hbase表
     * @param conn 链接
     * @param namespace 命名空间
     * @param tableName 表名
     * @param families 列族
     */
    public static void createTable(Connection conn,String namespace,String tableName,String... families){
        if (families.length<1){
            throw new IllegalArgumentException("请输入至少一个列族名...");
        }
        Admin admin;

        try {
            admin = conn.getAdmin();

            TableName tableNameHbase = TableName.valueOf(namespace, tableName);
            if (admin.tableExists(tableNameHbase)){
                System.out.println("要创建的表: "+namespace+":"+tableName+"已存在。。。");
                return;
            }

            //设置表描述对象
            TableDescriptorBuilder descriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameHbase);
            //指定列族
            for (String family : families) {
                ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                descriptorBuilder.setColumnFamily(familyDescriptor);
            }

            admin.createTable(descriptorBuilder.build());

        } catch (IOException e) {
            System.out.println("获取HbaseAdmin失败");
            throw new RuntimeException(e);
        }
    }

    /**
     * 向hbase中添加或更新数据
     * @param conn
     * @param nameSpace
     * @param tableName
     * @param rowKey
     * @param family
     * @param columns 字符串数组格式的列名
     * @param values 字符串数组格式的值
     * @throws IOException
     */
    public static void putRow(Connection conn,
                              String nameSpace,
                              String tableName,
                              String rowKey,
                              String family,
                              String[] columns,
                              String[] values) throws IOException {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf(nameSpace, tableName));
        //传入参数创建put对象
        Put put = new Put(rowKey.getBytes(StandardCharsets.UTF_8));

        for (int i = 0; i < columns.length; i++) {
            if (values[i]!=null){
                put.addColumn("info".getBytes(StandardCharsets.UTF_8),columns[i].getBytes(StandardCharsets.UTF_8),values[i].getBytes(StandardCharsets.UTF_8));
            }
        }

        table.put(put);
        table.close();
    }

    /**
     * 删除数据
     * @param conn
     * @param nameSpace
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void delRow(Connection conn,
                              String nameSpace,
                              String tableName,
                              String rowKey) throws IOException {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf(nameSpace, tableName));
        //传入参数创建delete对象
        Delete delete = new Delete(rowKey.getBytes(StandardCharsets.UTF_8));
        table.delete(delete);

        table.close();

    }


    //获取字典维度表建表语句
    public static String getBaseDicLookUpDDL() {
        return "CREATE TABLE dic (" +
                // 原子类型会自动识别为 hbase 的 RowKey,只能有一个原子类型. 列名随意
                " dic_code string," +
                // 声明列族,类型必须是 Row 类型, 嵌套的字段名是 hbase 中的列.
                " info ROW<dic_name string>," +
                " PRIMARY KEY (dic_code) NOT ENFORCED" +
                ")" + getHbaseDDL("edu_realtime:dim_base_dic");
    }

    //获取Hbase连接器相关参数
    public static String getHbaseDDL(String tableName) {
        return " with(" +
                " 'connector' = 'hbase-2.2'," +
                " 'zookeeper.quorum' = 'localhost:12181', " +
                " 'table-name' = '" + tableName + "', " +
                " 'lookup.async' = 'true', " +
                " 'lookup.cache' = 'PARTIAL'," +
                " 'lookup.partial-cache.max-rows' = '200'," +
                " 'lookup.partial-cache.expire-after-write' = '1 hour', " +
                " 'lookup.partial-cache.expire-after-access' = '1 hour'" +
                ")";
    }


    //获取hbase同步链接
    public static AsyncConnection getAsyncConnection(){
        try {
            Configuration conf = HBaseConfiguration.create();
            //配置zookeeper
            conf.set("hbase.zookeeper.quorum","localhost:12181,localhost:12182");
            AsyncConnection asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
            return asyncConnection;
        }catch (Exception e){
            throw new  RuntimeException();
        }
    }

    //关闭habse同步链接
    private static void closeAsyncConnection(Connection conn){
        if (conn != null&&!conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

        }
    }

    //异步方式从hbase中读取维度数据
    public static JSONObject getDimInfoFromHbaseAsync(AsyncConnection conn, String nameSpace, String tableName, String rowKey){
        AsyncTable<AdvancedScanResultConsumer> table = conn.getTable(TableName.valueOf(nameSpace, tableName));
        Get get = new Get(rowKey.getBytes(StandardCharsets.UTF_8));
        CompletableFuture<Result> resultCompletableFuture = table.get(get);

        JSONObject jsonObject = new JSONObject();
        try {
            Result result = resultCompletableFuture.get();
            Cell[] cells = result.rawCells();
            if (cells!=null&&cells.length>0){
                for (Cell cell : cells) {
                    jsonObject.put(Bytes.toString(CellUtil.cloneQualifier(cell)),Bytes.toString(CellUtil.cloneValue(cell)));
                }
                return jsonObject;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

}
